

# Shuffle 的本质是跨节点数据重组​​，是分布式计算中开销最大的操作之一

# MapReduce：
# 两阶段模型：强制将计算拆分为Map（数据映射/过滤）和Reduce（聚合输出）两个阶段，中间结果必须落盘存储。
# 仅支持Map/Reduce两类操作，复杂逻辑需串联多个Job
# 中间结果落盘，失败重启task，可靠性高，恢复慢
# 多进程模型：每个Task运行在独立JVM进程中，启动销毁开销大（约1秒/任务），资源利用率低。
# 强排序依赖：Shuffle阶段默认全排序，增加磁盘I/O和延迟。
# 超大规模，简单 ETL

# Spark：
# DAG驱动：将任务分解为有向无环图（DAG），优化执行路径，合并多阶段操作，减少冗余I/O。
# 提供map、filter、join等超100种算子，支持多步操作单Job完成
# 基于血缘记录生成步骤，节点故障按依赖关系重算丢失分区，避免完全恢复
# 内存优先：中间结果优先缓存于内存（RDD），迭代计算无需重复读写磁盘，仅Shuffle时落盘。
# 多线程模型：Task在Executor线程池内运行，复用线程降低启动开销（微秒级
# 占内存,数据缓存可能引发oom或频繁gc，shuffle 数据倾斜时容易失败，手动优化分区策略
# 低延迟，



#  dataFrame.cache() 来使用内存列式格式缓存表。然后 Spark SQL 将仅扫描所需的列，并自动调整压缩以最小化内存使用和 GC 压力。


# https://github.com/Devinterview-io/apache-spark-interview-questions
# https://github.com/cartershanklin/pyspark-cheatsheet



# rdd 是spark 中的数据抽象， 弹性分布式数据集，可以理解为一些数据，这些数据有，不可变和可并行操作的特点
# 有几个特性，结合弹性和不可变星，通过伸缩性可以处理故障恢复，有一些关键属性，分区列表，定义了rdd有哪些数据分别在哪
# 计算函数，每个rdd需要执行的计算逻辑，依赖列表，依赖哪些rdd
# 多个rdd 分布在不同节点并行计算，




#  ------------------------------------ rdd算子 ---------------------------------------------
# 转换算子（Transformations） 是惰性的，只记录计算逻辑，不立即执行。
rdd = sc.parallelize([1, 2, 3])
rdd.map(lambda x: x * 2)  # 输出: [2, 4, 6]

rdd.filter(lambda x: x > 1)  # 输出: [2, 3]

rdd.flatMap(lambda x: [x, x*10])  # 输出: [1, 10, 2, 20, 3, 30]

sc.parallelize([1, 1, 2]).distinct()  # 输出: [1, 2]

rdd1 = sc.parallelize([1, 2])
rdd2 = sc.parallelize([3, 4])
rdd1.union(rdd2)  # 输出: [1, 2, 3, 4]

rdd1.intersection(sc.parallelize([2, 3]))  # 输出: [2]

pair_rdd.reduceByKey(lambda a, b: a + b)  # 输出: [("a", 4), ("b", 2)] 键值对，会发生shuffle
reduce # 单一值会预聚合，然后把结果合并，不会发生shuffle

sc.parallelize([(3, "z"), (1, "a")]).sortByKey()  # 输出: [(1, "a"), (3, "z")]

rdd1 = sc.parallelize([(1, "A"), (2, "B")])
rdd2 = sc.parallelize([(1, "X"), (2, "Y")])
rdd1.join(rdd2)  # 输出: [(1, ("A", "X")), (2, ("B", "Y"))]

rdd.repartition(4)  # 将分区数改为 4



# --------------- 引发 shuffle 的操作 ----------------
# 重分区类操作​​
#     repartition()：调整分区数（例如从少分区扩增到多分区）。
# ​​ByKey 类聚合操作​​
#     groupByKey()：按 Key 分组（全量数据传输，无预聚合）。
#     reduceByKey()、aggregateByKey()：先局部聚合再全局聚合（减少数据传输）。
#     sortByKey()、countByKey()：需全局排序或统计。
# ​​连接类操作​​
#     join()、cogroup()：需将相同 Key 的数据拉取到同一节点。
#     leftOuterJoin、fullOuterJoin：外连接同样触发 Shuffle



# ------- 行动 rdd算子 ---------------
# 行动算子（Actions） 触发实际计算（如 collect(), count()）。
rdd.collect()  # 输出: [1, 2, 3]
rdd.count()  # 输出: 3
rdd.first()  # 输出: 1
rdd.take(2)  # 输出: [1, 2]
rdd.reduce(lambda a, b: a + b)  # 输出: 6
rdd.foreach(lambda x: print(x))

pair_rdd.aggregateByKey(0, lambda a, b: a + b, lambda a, b: a + b)
rdd.partitionBy(2, lambda x: x % 2)  # 按奇偶分区






#  ------------------------------------ 执行计划 ---------------------------------------------
# sparksession.sql(sql).explain() #  传  空 和 simple 表示只展示物理执行计划， 传 extended 表示展示逻辑和物理执行计划
#                                              formated 格式化物理执行计划
# hashaggregate 数据聚合
# exchange 表示 shuffle
# project 表示 列裁剪

# 还可以通过ui看（SQL)标签页看




#  ------------------------------------ 资源调优 ---------------------------------------------
# driver不参与具体运算，除非 调用 collect 将具体数据传给driver
# executor  一般 3-6 之间
# 单个 executor 给多少内存 还要看 yarn 中默认最大，最小 的内存，默认最小 1G 最大 8G



# 总内存 - 系统预留  剩下的 other 占 40%  用来存元数据和自定义数据结构

# 剩下 60%  storage 和 execution 平均分

# storage 存 cache、persist、广播变量
# execution 存 shuffle （join groupby reducebykey）

# storage内存    30%   --  广播变量 + 缓存/ executor
# execution内存  30%   --  每个 executor 核数 * （数据集大小 / 并行度)  sparksql默认shuffle并行度200,也就是task
                # 100G 数据；  每个 task 需要 500M内存， 如果一个 executor 有4个核 ， 那么此时每个exector 需要 2G 才够用
# other内存   40%      --  自定义数据结构 * 每个executor 核数
# 系统内存

# 上面加一起才是总内存数


#  ------------------------------------ rdd 和 df内存序列化 ---------------------------------------------
    # df 不需要考虑， rdd 需要考虑

# rdd 默认 内存缓存
# cache()  --基于-->  persist(缓存策略)  ：缓存策略 ，放磁盘， 放内存， 磁盘内存都放， 内存+备份，内存+磁盘备份
# cache 结合 使用 kryo序列化 占用空间更小


# df  cache 默认使用 缓存+磁盘
# 只需要修改 缓存级别为 内存+磁盘+序列化 MEMORY_AND_DISK_SER, df序列化方法自己选的不需要手动指定





#  ------------------------------------ 并行度 ---------------------------------------------
# cpu优化
# spark.default.parallelism
# rdd 默认并行度 由 join reducebykey parallelize 等转换决定的


# spark sql
# spark.sql.shuffle.partitions 默认200


# 并行度： 同时能执行 task 的数量，也就是spark集群同时有多少个核心可以使用
# 并行度较低，也就是task数量少， 数据分配较大（一个task处理的数据量太大），容易导致cpu线程挂起， 一个task 5G 两个task10G，此时总内存不到10G那么只能跑一个任务
# 并行度过高，数据太分散，调度开销更大

# 官方推荐：并行度（task数）设置成集群并发度（vcore）的 2 到 3 倍




#  ------------------------------------ sql语法优化 ---------------------------------------------
# RBO
# 谓词下推 ： 读表时过滤条，提前数据减少
# 注意：  select * from a  left join  b on xx=xx where a.yy>0   这种条件再where中的可能导致提前过滤（因为索引下推的原因）

# 列裁剪 和 常量替换



#  ------------------------------------ CBO物理执行计划后选一个代价相对较小 ---------------------------------------------
# 关注 join 的优化
# ANALYZE  TABLE  xxx表  COMPUTE STATISTICS




#  ------------------------------------ 广播join ---------------------------------------------
# spark 中如果一张表足够小，会先缓存到内存，将小标聚合到 driver 再 广播 到各个大表分区中，大表join时就是基于本地的表，避免shuffle

# 默认10M

# 效果等同
# BROADCASTJOIN
# BROADCAST
# MAPJOIN 

# select /*+ BROADCASTJOIN(SC)*/    # SC 时小表，需要明确指定
# from sc join csc on xxxx






#  ------------------------------------ SMBjoin ---------------------------------------------
# 处理两张大表join  （sort  merge bucket）

# 必要条件：
# 1. 两表分桶，数量必须相同
# 2. 两边join 时 join列 == 排序列 == 分桶列

# 可以将两表都建一张中间分桶表

# bucketby(num=x, colname="xx").sortby("xx").save()





#  ------------------------------------ 数据倾斜 ---------------------------------------------

# 单表数据清洗： groupby
# spark 会提前预聚合，下游可以直接读取预聚合的结果，避免大量数据汇集到一起
# 加盐： 就是给大key 加上后最，让他分散，聚合完了再吧key 还原回去



# 把 大key 部分拆分出来，分成2表，打散（加盐）
# 此时需要将两一张表 扩容成多分，大key打散成几分， 这张表就得膨胀几倍





#  ------------------------------------ job优化 ---------------------------------------------
# 如果用rdd 建议用 reducebykey 和 aggregatebykey 替换 groupby ，因为他们会本地预聚合

# 读取小文件优化：
# spark.sql.files.maxPartitionBytes=128M  将小文件读到一个分区中
# spark.files.openCostInBytes = 4194304  # 打开文件开销， 默认 4M


distribute by 

# spark.sql.shuffle.partitions  设置合理的分区，避免过多（小文件）和过少（数据倾斜）